package com.example.rocket.consumer;

import com.example.rocket.annotation.NbMQMessageListener;
import com.example.rocket.core.ConsumeMode;
import com.example.rocket.core.NbMsg;
import com.example.rocket.core.SelectorType;
import com.example.rocket.utils.JSONUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.Objects;

@Slf4j
@Data
public abstract class AbstractNbMQConsumer implements InitializingBean, SmartLifecycle, ApplicationContextAware, DisposableBean {

    protected ApplicationContext applicationContext;

    protected String nameServer;

    protected String topic;

    /**
     * 等价于 TAG
     */
    protected String selectorExpression;

    /**
     * 消费者Group
     */
    protected String consumerGroup;

    /**
     * accessKey
     */
    protected String accessKey;

    /**
     * secretKey
     */
    protected String secretKey;

    /**
     * 消费模式
     */
    protected ConsumeMode consumeMode;

    /**
     * TAG筛选模式
     */
    protected SelectorType selectorType;

    /**
     * 消息订阅模式
     */
    protected MessageModel messageModel;

    /**
     * 消费者最大线程数
     */
    protected int consumeThreadMax = 64;

    protected int delayLevelWhenNextConsume = 0;

    /**
     * 消费超时时间
     */
    protected long consumeTimeout;

    protected String charset = "UTF-8";

    protected AccessChannel accessChannel = AccessChannel.LOCAL;


    /**
     * 消息监听器
     */
    protected NbMqListener nbMqListener;

    /**
     * 注解
     */
    protected NbMQMessageListener nbMQMessageListener;


    /**
     * 是否运行
     */
    protected boolean running;

    /**
     * 消息装换类型
     */
    protected Type messageType;


    @Override
    public void afterPropertiesSet() throws Exception {
        log.trace("默认监听器容器初始化");
        initNbMQPushConsumer();
        this.messageType = getMessageType();
        log.trace("===>"+this.messageType);
    }

    @Override
    public void stop(Runnable callback) {
        log.trace("默认===>stop-Runnable");
        stop();
        callback.run();
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }


    /**
     * 填充consumer其他属性
     * @param data
     */
    public void setOtherProperties(NbMQMessageListener data){
        this.nbMQMessageListener = data;
        this.topic = data.topic();
        this.selectorExpression = data.selectorExpression();
        this.consumerGroup = data.consumerGroup();
        this.consumeMode = data.consumeMode();
        this.consumeThreadMax = data.consumeThreadMax();
        this.messageModel = data.messageModel();
        this.selectorType = data.selectorType();
        this.selectorExpression = data.selectorExpression();
        this.consumeTimeout = data.consumeTimeout();
        this.accessKey = data.accessKey();
        this.secretKey = data.secretKey();
    }


    /**
     * 初始化 MQ 消息监听器
     * @throws MQClientException
     */
    protected abstract void initNbMQPushConsumer() throws MQClientException ;

    /**
     * 消息处理器
     * @param
     */
    protected void handleMessage(NbMsg msg) {
        log.trace("===>"+"默认handleMessage");
        if (nbMqListener != null) {
            nbMqListener.onMessage(doConvertMessage(msg));
        }
    }


    /**
     * 消息转换器：更具消费者实现的接口传入的泛型，将消息转换成对应的类型
     * @param msg
     * @return
     */
    protected  Object doConvertMessage(NbMsg msg){
        log.trace("===>"+"默认doConvertMessage");
        if (Objects.equals(messageType, NbMsg.class)) {
            return msg;
        }else{
            String bodySty = new String(msg.getBody(), Charset.forName(charset));
            if (Objects.equals(messageType, String.class)) {
                return bodySty;
            }else{
                try {
                    return JSONUtils.readValue(msg.getBody(),  (Class<?>)messageType);
                } catch (Exception e) {
                    log.error("convert failed. str:{}, msgType:{}", bodySty, messageType);
                    throw new RuntimeException("cannot convert message to " + messageType, e);
                }
            }
        }
    }

    /**
     * 获取消息类型：实际上就是获取消费者监听器实现的接口传入的泛型，
     * @return
     */
    protected Type getMessageType() {
        log.trace("===>"+"默认获取消息类型");
        Class<?> targetClass;
        if (nbMqListener != null) {
            targetClass = AopProxyUtils.ultimateTargetClass(nbMqListener);
        } else {
            //TODO 其他类型
            targetClass = AopProxyUtils.ultimateTargetClass(nbMqListener);
        }
        Type matchedGenericInterface = null;
        while (Objects.nonNull(targetClass)) {
            Type[] interfaces = targetClass.getGenericInterfaces();
            if (Objects.nonNull(interfaces)) {
                for (Type type : interfaces) {
                    if (type instanceof ParameterizedType && (Objects.equals(((ParameterizedType) type).getRawType(), NbMqListener.class) )) {
                        matchedGenericInterface = type;
                        break;
                    }
                }
            }
            targetClass = targetClass.getSuperclass();
        }
        if (Objects.isNull(matchedGenericInterface)) {
            return Object.class;
        }

        Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();
        if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
            return actualTypeArguments[0];
        }
        return Object.class;
    }


    @Override
    public String toString() {
        return "DefaultRocketMQListenerContainer{" +
                "consumerGroup='" + consumerGroup + '\'' +
                ", nameServer='" + nameServer + '\'' +
                ", topic='" + topic + '\'' +
                ", selectorExpression='" + selectorExpression + '\'' +
                '}';
    }


}
